from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from sqlalchemy import create_engine, text
import pandas as pd
import numpy as np
import logging
from typing import List, Optional
import uvicorn

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# 数据库连接配置
DB_URI = "mysql+pymysql://root:1234@localhost:3306/two"

# 创建 FastAPI 应用
app = FastAPI(
    title="客户金融数据分析 API",
    description="提供客户金融数据分析和预测服务",
    version="1.0.0"
)

# 数据模型定义
class CustomerFinancialData(BaseModel):
    age: int
    occupation: str
    industry: str
    monthly_income: float
    credit_score: int
    overdue_count: int
    debt_payment: float
    total_assets: float
    total_liabilities: float
    liquid_assets: float
    credit_card_utilization: float

class FinancialHealthResponse(BaseModel):
    customer_id: int
    financial_health_score: float
    risk_category: str
    recommendations: List[str]

# 数据库连接工具
def get_db_connection():
    """创建并返回数据库连接"""
    try:
        engine = create_engine(DB_URI)
        connection = engine.connect()
        logger.info("数据库连接成功")
        return connection
    except Exception as e:
        logger.error(f"数据库连接失败: {e}")
        raise HTTPException(
            status_code=500,
            detail=f"数据库连接错误: {str(e)}"
        )

# 数据加载函数
# 数据加载函数（修复 Decimal 问题）
def load_data(limit: int = None) -> pd.DataFrame:
    """从数据库加载客户金融数据"""
    try:
        with get_db_connection() as conn:
            # 在查询中直接转换 Decimal 为 FLOAT
            query = text(f"""
                SELECT
                    c.customer_id, c.age, c.occupation, c.industry,
                    CAST(f.monthly_income AS FLOAT) AS monthly_income,
                    CAST(f.credit_score AS FLOAT) AS credit_score,
                    f.overdue_count,
                    CAST(f.debt_payment AS FLOAT) AS debt_payment,
                    CAST(f.total_assets AS FLOAT) AS total_assets,
                    CAST(f.total_liabilities AS FLOAT) AS total_liabilities,
                    CAST(f.liquid_assets AS FLOAT) AS liquid_assets,
                    CAST(f.credit_card_utilization AS FLOAT) AS credit_card_utilization
                FROM customer_basic_info c
                JOIN financial_data f ON c.customer_id = f.customer_id
                {'LIMIT :limit' if limit else ''}
            """)

            params = {"limit": limit} if limit else {}
            result = conn.execute(query, params)

            rows = result.fetchall()
            columns = result.keys()
            df = pd.DataFrame(rows, columns=columns)

            logger.info(f"成功加载 {len(df)} 条记录")
            return df
    except Exception as e:
        logger.error(f"数据加载失败: {e}")
        raise HTTPException(
            status_code=500,
            detail=f"数据加载错误: {str(e)}"
        )

# 数据预处理函数
def preprocess_data(df: pd.DataFrame) -> pd.DataFrame:
    """对金融数据进行预处理"""
    try:
        # 1. 处理缺失值
        df.fillna({
            'monthly_income': df['monthly_income'].median(),
            'credit_score': df['credit_score'].median(),
            'overdue_count': 0,
            'debt_payment': 0,
            'credit_card_utilization': df['credit_card_utilization'].mean()
        }, inplace=True)

        # 2. 计算衍生特征
        df['debt_to_income_ratio'] = df['debt_payment'] / df['monthly_income']
        df['net_worth'] = df['total_assets'] - df['total_liabilities']
        df['liquidity_ratio'] = df['liquid_assets'] / df['total_liabilities'].replace(0, 1)

        # 3. 标准化数值特征
        numeric_cols = [
            'monthly_income', 'credit_score', 'debt_payment',
            'total_assets', 'total_liabilities', 'liquid_assets',
            'debt_to_income_ratio', 'net_worth', 'liquidity_ratio'
        ]

        for col in numeric_cols:
            df[f'scaled_{col}'] = (df[col] - df[col].mean()) / df[col].std()

        logger.info("数据预处理完成")
        return df
    except Exception as e:
        logger.error(f"数据预处理失败: {e}")
        raise HTTPException(
            status_code=500,
            detail=f"数据处理错误: {str(e)}"
        )

# 金融健康分析模型
def analyze_financial_health(df: pd.DataFrame, customer_id: int) -> dict:
    """分析客户金融健康状况"""
    try:
        if customer_id not in df['customer_id'].values:
            raise ValueError(f"客户ID {customer_id} 不存在")

        customer_data = df[df['customer_id'] == customer_id].iloc[0]

        # 计算金融健康评分 (0-100)
        score = 0

        # 收入因素 (30%)
        income_score = min(100, customer_data['monthly_income'] / 10000 * 20)
        score += income_score * 0.3

        # 债务因素 (25%)
        debt_ratio = customer_data['debt_to_income_ratio']
        debt_score = 100 * (1 - min(1, debt_ratio / 0.4))  # 债务收入比超过40%扣分
        score += debt_score * 0.25

        # 信用因素 (20%)
        credit_score = customer_data['credit_score']
        score += min(100, credit_score / 850 * 100) * 0.2

        # 流动性因素 (15%)
        liquidity_score = min(100, customer_data['liquidity_ratio'] * 50)  # 流动性比率2以上得满分
        score += liquidity_score * 0.15

        # 逾期因素 (10%)
        overdue_count = customer_data['overdue_count']
        overdue_score = max(0, 100 - (overdue_count * 20))  # 每有一次逾期扣20分
        score += overdue_score * 0.1

        # 确定风险类别
        if score >= 80:
            risk_category = "低风险"
            recommendations = [
                "继续保持良好的财务管理习惯",
                "考虑投资多元化以增加财富"
            ]
        elif score >= 60:
            risk_category = "中低风险"
            recommendations = [
                "监控债务水平，避免增加新债务",
                "建立应急基金"
            ]
        elif score >= 40:
            risk_category = "中风险"
            recommendations = [
                "优先偿还高利率债务",
                "减少不必要的开支",
                "提高信用评分"
            ]
        else:
            risk_category = "高风险"
            recommendations = [
                "立即制定债务偿还计划",
                "寻求专业财务咨询",
                "考虑债务重组或合并"
            ]

        return {
            "customer_id": customer_id,
            "financial_health_score": round(score, 2),
            "risk_category": risk_category,
            "recommendations": recommendations
        }

    except Exception as e:
        logger.error(f"金融健康分析失败: {e}")
        raise HTTPException(
            status_code=400,
            detail=f"分析错误: {str(e)}"
        )

# API 路由定义
@app.get("/", tags=["Root"])
def read_root():
    """API 根端点"""
    return {
        "message": "欢迎使用客户金融数据分析 API",
        "endpoints": [
            {"path": "/customers", "method": "GET", "description": "获取客户数据"},
            {"path": "/analyze/{customer_id}", "method": "GET", "description": "分析客户金融健康状况"},
            {"path": "/health", "method": "GET", "description": "服务健康检查"}
        ]
    }

@app.get("/health", tags=["Utility"])
def health_check():
    """服务健康检查"""
    try:
        # 测试数据库连接
        with get_db_connection() as conn:
            conn.execute(text("SELECT 1"))
        return {"status": "healthy", "database": "connected"}
    except Exception as e:
        return {"status": "unhealthy", "error": str(e)}

@app.get("/customers", response_model=List[CustomerFinancialData], tags=["Data"])
def get_customers(limit: int = 100):
    """获取客户金融数据"""
    try:
        df = load_data(limit)
        # 转换为 Pydantic 模型
        return df.to_dict(orient='records')
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/analyze/{customer_id}", response_model=FinancialHealthResponse, tags=["Analysis"])
def analyze_customer(customer_id: int):
    """分析指定客户的金融健康状况"""
    try:
        # 加载并预处理数据
        df = load_data()
        processed_df = preprocess_data(df)

        # 进行分析
        result = analyze_financial_health(processed_df, customer_id)
        return result
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

@app.post("/predict", tags=["Analysis"])
def predict_financial_health(data: CustomerFinancialData):
    """基于输入数据预测金融健康状况"""
    try:
        # 将输入数据转换为DataFrame
        input_df = pd.DataFrame([data.dict()])

        # 预处理
        processed_df = preprocess_data(input_df)

        # 模拟分析 (实际应用中应使用训练好的模型)
        # 这里使用第一个"客户"的ID作为占位符
        customer_id = 9999
        processed_df['customer_id'] = customer_id

        result = analyze_financial_health(processed_df, customer_id)

        # 移除客户ID
        result.pop('customer_id')

        return result
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

# 主程序入口
if __name__ == "__main__":
    # 启动 API 服务
    uvicorn.run("detector:app",host='127.0.0.1',port=8084,reload=True,workers=1)
